package com.teamsun.mqtt.service;

import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttConnAckVariableHeader;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;


import org.apache.ibatis.session.SqlSession;

import com.teamsun.mqtt.db.DBSessionFactory;
import com.teamsun.mqtt.entity.vo.UserValidate;

/**
 * 
 * @author acer
 *
 */
public class UserPwdQueCall implements  Callable<Void>{

	List<UserValidate> userValidates;
	
	ConcurrentHashMap<String, Channel> str2channel;

	ConcurrentHashMap<Channel, String> channel2str;
	
	
	
	final  String validateSql = "com.teamsun.mqtt.entity.UserMapper.validateLogin";//查询
	
	static UserPwdQueCall userPwdQueCall;
	
	public static UserPwdQueCall  getInstance(){
		
		if(userPwdQueCall==null)
			userPwdQueCall=new UserPwdQueCall();
		
		return   userPwdQueCall;
	}
	
	private UserPwdQueCall() {
	
	}


	public UserPwdQueCall(List<UserValidate> userValidates,
			ConcurrentHashMap<String, Channel> str2channel,
			ConcurrentHashMap<Channel, String> channel2str) {
		super();
		this.userValidates = userValidates;
		this.str2channel = str2channel;
		this.channel2str = channel2str;
	}




/**
 * 批量查询数据库验证用户名密码
 */
	@SuppressWarnings("unchecked")
	@Override
	public Void call() throws Exception {

		
		 if(userValidates==null)
			 throw new NullPointerException();
		 
		 
		 Map<String, Object> map=new HashMap<String, Object>();
		 map.put("uses", userValidates);
		 
		
		 SqlSession session= DBSessionFactory.getSqlSession();
		 List<?> results=(List<?>)session.selectList(validateSql,map);
		 
		 session.close();
		 
		 map.clear();
		 
		 MqttFixedHeader fixedHeader = new MqttFixedHeader(
					MqttMessageType.CONNACK, false, MqttQoS.AT_LEAST_ONCE, false,0);
		 
		 MqttConnAckVariableHeader connectVariableHeader = null;
	
			 for(Object object:results){
				 if(object instanceof Map){
					 map.putAll((Map<String,Long>)object);
				 }
			 }
		 
			 for(UserValidate validate:userValidates){
				 
				 Channel channel=validate.getChannel();
				 String key=validate.getUsername();
				 if(map.containsKey(key)){
					 Long  l= (Long) map.get(key);
					 if(l>0)
					 connectVariableHeader= new MqttConnAckVariableHeader(
								MqttConnectReturnCode.CONNECTION_ACCEPTED, false);
					 else
						 connectVariableHeader= new MqttConnAckVariableHeader(
									MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false);
					 
					 
					 MqttConnAckMessage ackMessage = new MqttConnAckMessage(fixedHeader,
								connectVariableHeader);
					 channel.writeAndFlush(ackMessage);
					 
					 
					 str2channel.put(validate.getIden(), channel);
						channel2str.put(channel, validate.getIden());
						 
				 }
				
					
					
			 }
			 userValidates=null;
			
		return null;

	}

public List<UserValidate> getUserValidates() {
	return userValidates;
}

public void setUserValidates(List<UserValidate> userValidates) {
	this.userValidates = userValidates;
}

public ConcurrentHashMap<String, Channel> getStr2channel() {
	return str2channel;
}

public void setStr2channel(ConcurrentHashMap<String, Channel> str2channel) {
	this.str2channel = str2channel;
}

public ConcurrentHashMap<Channel, String> getChannel2str() {
	return channel2str;
}

public void setChannel2str(ConcurrentHashMap<Channel, String> channel2str) {
	this.channel2str = channel2str;
}

public String getValidateSql() {
	return validateSql;
}
	
	

}
